# @Project: diablo
# @File: insert_history_stock_price
# @Time: 2023/8/31 19:39
# @User: Jolly
# @Email: jolly@007vin.com

"""
查询近3天订单计算历史库存定价
"""

import datetime
import time
import traceback
import threading
from queue import Queue

import pymysql
# from datetime import datetime

from pymongo import MongoClient, InsertOne, UpdateOne
from dbutils.pooled_db import PooledDB
from pymongo.errors import ConnectionFailure


class InsertHistoryStockPrice:
    """
    插入历史订单库存价格
    """

    def __init__(self):

        # start_time = time.time()

        # 初始化数据库连接
        # 2.0生产库
        self.diablo_conn_2_0 = self.get_mongo_connection("root", "aa123456", "192.168.191.175", "27017")

        # 正式mysql
        self.mysql_conn = self.get_mysql_connection()

        # print("========= 数据插入,共耗时:{}'s =========".format(round(time.time() - start_time, 3)))

    def get_mongo_connection(self, username, password, host, port):
        try:
            # 创建 MongoClient
            client = MongoClient(f"mongodb://{username}:{password}@{host}:{port}",
                                 maxPoolSize=150,
                                 minPoolSize=1,
                                 maxIdleTimeMS=30000,
                                 waitQueueTimeoutMS=2000)

            # 检测连接是否可用，如果连接不可用则抛出异常
            if not client:
                raise ConnectionFailure("从MongoDB连接池中获取连接失败！")

            return client

        except ConnectionFailure as e:
            # 处理连接失败的异常
            print(f"连接MongoDB失败: {e}")
            return None

    def get_mysql_connection(self):
        maxconnections = 50  # 最大连接数
        # 本地
        # pool = PooledDB(
        #     pymysql,
        #     maxconnections,
        #     host='127.0.0.1',
        #     user='root',
        #     port=3306,
        #     passwd='12345678',
        #     db='users',
        #     use_unicode=True)

        # 线上
        pool = PooledDB(
            pymysql,
            maxconnections,
            host='192.168.191.21',
            user='users',
            port=3306,
            passwd='uSR@ll7GnJDopei',
            db='users',
            use_unicode=True)

        # 测试
        # pool = PooledDB(
        #     pymysql,
        #     maxconnections,
        #     host='192.168.191.47',
        #     user='users',
        #     port=3306,
        #     passwd='uSR@#H^&JD',
        #     db='users',
        #     use_unicode=True)

        return pool

    def get_day_time_format(self):
        """
        获取今天和三天前的时间
        """
        # 获取当前日期
        today = datetime.datetime.now()

        # 将当前日期的时间部分设置为23:59:59
        today = today.replace(hour=23, minute=59, second=59)

        # 计算三天前的日期
        three_days_ago = today - datetime.timedelta(days=3)

        # 将三天前的日期的时间部分设置为00:00:00
        three_days_ago = three_days_ago.replace(hour=0, minute=0, second=0)

        # 格式化日期和时间为字符串
        today_str = today.strftime('%Y-%m-%d %H:%M:%S')
        three_days_ago_str = three_days_ago.strftime('%Y-%m-%d %H:%M:%S')

        return today_str, three_days_ago_str

    def get_last_3_day_orders(self, merchant_id):
        """
        获取获取3天订单信息
        """

        connection = self.diablo_conn_2_0

        today_str, three_days_ago_str = self.get_day_time_format()

        inquiry_data = []

        try:
            if connection is not None:
                # 执行数据库操作
                db = connection["ddll"]
                conn = db["ddl_inquiry"]

                filter_dict = {
                    "merchant_id": merchant_id,
                    "create_time": {
                        "$gt": three_days_ago_str,
                        "$lt": today_str
                    }
                }
                docs = conn.find(filter_dict)

                if conn.count_documents(filter_dict) == 0:
                    return [], True

                inquiry_data = [inquiry for inquiry in docs]

                # return data, False

        except Exception as e:
            print("获取三日询价单数据失败 - mongo：", e)
        finally:
            # 将连接放回连接池
            connection.close()

        previous = inquiry_data[0]
        for row in inquiry_data:
            yield previous, False
            previous = row
        yield previous, True

    def get_merchant_data(self):
        """
        获取所有商家信息，然后遍历写入看板数据
        """
        mysql_conn = self.mysql_conn.connection()

        try:
            cursor = mysql_conn.cursor()

            # 查询商家id，加入列表返回
            merchant_ids = []
            merchant_sql = f"select DISTINCT(merchant_id) merchant_id from diablo_merchant "
            cursor.execute(merchant_sql)
            merchant_data = cursor.fetchall()

            for merchant in merchant_data:
                merchant_ids.append(merchant[0])

            return merchant_ids

        except Exception as e:
            print(f"获取商家信息失败 {e}")
        finally:
            mysql_conn.close()

    def insert_recommend_stock_price_data(self, insert_list):
        """
        插入推荐价格记录，对于同一商家同一客户的同一品质进行覆盖操作，仅保留一条记录
        """
        # 获取mongo数据库连接
        connection = self.diablo_conn_2_0

        update_list = []

        try:
            if connection is not None:
                # 执行数据库操作
                db = connection["ddll"]
                conn = db["stock_recommend_price_history"]  # 历史定价推荐记录表

                # 查看是否存在记录，不存在则写入

                for data in insert_list:

                    filter_dict = {
                        "merchant_id": data.get("merchant_id"),
                        "identify": data.get("identify"),
                        "customer": data.get("customer")
                    }

                    update_dict = {
                        "$set": data
                    }

                    update_list.append(UpdateOne(filter_dict, update_dict, upsert=True))

                if len(update_list) > 0:
                    conn.bulk_write(update_list)

        except Exception as e:
            print("写入历史推荐定价数据失败 - mongo：", e)
            print(traceback.format_exc())
        finally:
            # 将连接放回连接池
            connection.close()

    def insert_stock_price_data(self, *args):
        """
        写入库存价格数据
        """

        inquiry_data = args[0]

        # 获取今天开始时间，比较询价单在这之前的，就是操作更新，在这之后的则是操作插入

        today_start = datetime.datetime.now().strftime('%Y-%m-%d') + " 00:00:00"

        # 将时间字符串转换为日期时间对象
        today_start_obj = datetime.datetime.strptime(today_start, "%Y-%m-%d %H:%M:%S")

        # 拼接整合待写入译码数据
        formatted_datetime = datetime.datetime.now().strftime('%Y-%m-%d')

        # for data in inquiry_data:

        operate_list = []

        # 待确认之前的都舍弃
        inquiry_status = inquiry_data.get("inquiry_status")
        if inquiry_status < 5:
            return

        create_time = inquiry_data.get("create_time")

        create_time_obj = datetime.datetime.strptime(create_time, "%Y-%m-%d %H:%M:%S")

        part_list = inquiry_data.get("part_list")

        if not part_list or len(part_list) == 0:
            return

        insert_list = []

        for part in part_list:

            oe = part.get("oe")

            if not oe or oe == "":
                continue

            quality_price = part.get("quality_price")

            if not quality_price or len(quality_price) == 0:
                continue

            for stock in quality_price:
                identify = stock.get("identify", "")
                quality = stock.get("quality_name")

                if not quality or quality == "":
                    continue

                quality = quality.split(")")[1] if ")" in quality else quality
                brand = stock.get("ori_brand", "")
                price = stock.get("quality_price", 0)

                insert_dict = {
                    "merchant_id": inquiry_data.get("merchant_id"),  # 商家
                    "inquiry_id": inquiry_data.get("inquiry_id"),  # 订单id
                    "customer": inquiry_data.get("inquiry_user").get("repair_shop_id"),  # 客户
                    "oe": oe,  # oe
                    "identify": identify,
                    "quality": quality,  # 品质
                    "brand": brand,  # 品牌
                    "order_status": 1 if inquiry_data.get("inquiry_status") == 5 else 2,  # 订单状态 1未成交 2已成交
                    "datetime": formatted_datetime,  # 时间
                    "price": price  # 价格
                }

                insert_list.append(insert_dict)

        # 写入历史推荐定价数据
        self.insert_recommend_stock_price_data(insert_list)

        if create_time_obj < today_start_obj:
            # 今天之前的订单，更新操作

            for stock_data in insert_list:

                # 过滤条件
                filter_dict = {
                    "merchant_id": stock_data.get("merchant_id"),
                    "inquiry_id": stock_data.get("inquiry_id"),
                    "customer": stock_data.get("customer"),
                    "identify": stock_data.get("identify")
                }

                update_dict = {'$set': inquiry_data}

                operate_list.append(UpdateOne(filter_dict, update_dict))
        else:
            # 今天的订单，插入操作

            for stock_data in insert_list:
                operate_list.append(InsertOne(stock_data))

        # 获取mongo数据库连接
        connection = self.diablo_conn_2_0

        try:
            if connection is not None:
                # 执行数据库操作
                db = connection["ddll"]
                conn = db["stock_price_history"]  # 历史定价记录表

                if len(operate_list) > 0:
                    conn.bulk_write(operate_list)

        except Exception as e:
            print("写入历史定价数据失败 - mongo：", e)
            print(traceback.format_exc())
        finally:
            # 将连接放回连接池
            connection.close()


if __name__ == "__main__":

    start_time = time.time()

    service = InsertHistoryStockPrice()

    # 获取商家id列表
    merchant_ids = service.get_merchant_data()
    if len(merchant_ids) == 0:
        print("未查询到商家id信息")

    for merchant_id in merchant_ids:

        q = Queue(maxsize=120)  # 设定最大队列数和线程数

        count = 1

        for row, last in service.get_last_3_day_orders(merchant_id):

            if row is False:
                continue

            t = threading.Thread(target=service.insert_stock_price_data, args=(row,))
            q.put(t)
            if (q.full() == True) or last:
                thread_list = []
                while q.empty() == False:
                    t = q.get()
                    thread_list.append(t)
                    t.start()
                for t in thread_list:
                    t.join()

            print("当前写入数：", count, " 商家为：", merchant_id)

            if last:
                break

            count += 1

        # inquiry_data, check = service.get_last_3_day_orders(merchant_id)
        # if check:
        #     print(f"商家id{merchant_id}三日内无询价单数据")
        # else:

            # 写入价格数据
            # service.insert_stock_price_data(inquiry_data)

    print("========= 数据插入,共耗时:{}'s =========".format(round(time.time() - start_time, 3)))



